package mspbots.hubspot.client.core.contacts;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.extern.log4j.Log4j2;
import mspbots.hubspot.client.core.AbstractToolsUtil;
import mspbots.hubspot.client.core.BaseApiService;
import mspbots.hubspot.common.*;
import org.springframework.cloud.client.loadbalancer.reactive.ReactorLoadBalancerExchangeFilterFunction;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * data.connector.hubspot.core.contacts.ContactService
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/3/31
 */
@Log4j2
@Service
public class ContactsServiceImpl extends AbstractToolsUtil implements BaseApiService {

    private final WebClient webClient;

    public ContactsServiceImpl(WebClient.Builder builder,
                               ReactorLoadBalancerExchangeFilterFunction filterFunction) {
        this.webClient = builder.baseUrl("http://hub-spot-data/")
                .filter(filterFunction)
                .build();
    }

    @Override
    public Flux<JsonNode> search(Long tenantId, QueryBody queryBody) {
        return this.hubSpotClient.fetch("/crm/v3/objects/contacts/search", queryBody)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    @Override
    public Flux<JsonNode> properties(Long tenantId) {
        return this.hubSpotClient.get("/crm/v3/properties/contacts", null)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId))
                .flatMapMany(jsonNode -> Flux.fromIterable(jsonNode.withArray("results")));
    }

    public Mono<JsonNode> associationsCreate(Long tenantId,Long contactId,Long ticketId) {
        String path = "/crm/v3/objects/contacts/" + contactId + "/associations/tickets/" + ticketId + "/contact_to_ticket";
        return this.hubSpotClient.put(path, null)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    /**
     * Create a batch of contacts
     * <p>
     * Create a batch of contacts. This follows the same rules as creating an individual object.
     *
     * @link https://developers.hubspot.com/docs/api/crm/contacts
     */
    public Flux<Object> batchDelete(Long tenantId, List<String> emails) {
        return this.webClient.post()
                .uri(uriBuilder -> uriBuilder.path("/contacts/{tenantId}/search").build(tenantId))
                .bodyValue(emails)
                .retrieve()
                .bodyToFlux(JsonNode.class)
                .map(jsonNode -> jsonNode.findValue("otherId").longValue())
                .flatMap(contactId -> this.hubSpotClient.delete("/crm/v3/objects/contacts/" + contactId)
                        .defaultIfEmpty(Map.of("result", contactId)))
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    public Flux<JsonNode> propertiesBatchUpdate(Long tenantId, List<ContactPropertiesBody> contactPropertiesBodies) {
        List<Mono<JsonNode>> monoList = contactPropertiesBodies.parallelStream()
                .map(contactPropertiesBody -> this.hubSpotClient
                        .patch("/crm/v3/properties/contacts/" + contactPropertiesBody.getName(),
                                contactPropertiesBody)).collect(Collectors.toList());
        return Flux.concat(monoList).subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    /**
     * Create a batch of contacts
     * <p>
     * Create a batch of contacts. This follows the same rules as creating an individual object.
     *
     * @link https://developers.hubspot.com/docs/api/crm/contacts
     */
    public Mono<JsonNode> batchCreate(Long tenantId, ContactCreateBody createBody) {
        return this.hubSpotClient.post("/crm/v3/objects/contacts/batch/create", createBody)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    public Flux<JsonNode> batchUpdate(Long tenantId, List<ContactUpdateBody> updateBodies) {
        log.debug("Update tenant [{}], body [{}]", tenantId, updateBodies);
        List<Flux<JsonNode>> monoList = updateBodies.parallelStream()
                .filter(updateBody -> updateBody.getEmails().size() > 0)
                .map(updateBody -> this.webClient.post()
                        .uri(uriBuilder -> uriBuilder.path("/contacts/{tenantId}/search").build(tenantId))
                        .bodyValue(updateBody.getEmails())
                        .retrieve().bodyToFlux(JsonNode.class)
                        .collectList()
                        .filter(list -> list.size() > 0)
                        .flatMapMany(contactsList -> {
                            List<Map<String, Object>> inputs = new ArrayList<>();
                            contactsList.forEach(entity -> inputs.add(Map.of("properties", updateBody.getProperties(),
                                    "id", entity.findValue("otherId").asLong(0))));
                            List<Mono<JsonNode>> monoPageList = new ArrayList<>();
                            int page = (inputs.size() / 10) + 1;
                            for (int x = 0; x < page; x++) {
                                monoPageList.add(this.hubSpotClient.post("/crm/v3/objects/contacts/batch/update",
                                        Map.of("inputs", inputs.stream().skip(x * 10).limit(10)
                                                .collect(Collectors.toList())))
                                        .subscriberContext(ctx -> ctx.put("tenantId", tenantId)));
                                log.info("Update page [{}], skip [{}]", x, x * 10);
                            }
                            return Flux.concat(monoPageList);
                        })).collect(Collectors.toList());

        return Flux.concat(monoList);
    }

    @Override
    public Mono<Void> syncProperties(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault("CONTACTS-ENTITY-PROPERTIES");
        return this.properties(tenantId)
                .subscriberContext(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(result -> this.syncPropertiesResult(tenantId,
                        resultSync.getRequestId(), result))
                .then();
    }

    private void syncPropertiesResult(Long tenantId, UUID id, JsonNode resultNode) {
        this.sendEntity(RabbitKeys.PROPERTIES_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(id).action("Contacts")
                .entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    @Override
    public Mono<Void> syncData(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault("CONTACTS-SYNC");
        return this.dateCacheFilterContacts(tenantId)
                .subscriberContext(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(result -> this.syncResult(tenantId, resultSync.getRequestId(), result))
                .then();
    }

    private void syncResult(Long tenantId, UUID id, JsonNode resultNode) {
        Long entityId = Long.parseLong(resultNode.get("id").textValue());
        log.debug("Contacts result id {}", resultNode.get("id").textValue());
        this.sendEntity(RabbitKeys.CONTACTS_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(id).id(entityId)
                .action("none").entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    private Flux<JsonNode> dateCacheFilterContacts(Long tenantId) {
        String redisAnchorKey = RedisKeys.CONTACTS_ANCHOR_REDIS_KEY_PREFIX + tenantId;
        return super.readAnchorTime(redisAnchorKey)
                .flatMapMany(syncTime -> this.search(tenantId,
                        QueryBody.withFilter(FilterExpression.builder()
                                .operator("GT").propertyName("lastmodifieddate")
                                .value(syncTime.toString()).build())))
                .doOnNext(user -> super.recordAnchorTime(redisAnchorKey,
                        ZonedDateTime.parse(user.get("updatedAt").textValue(), DateTimeFormatter.ISO_DATE_TIME)
                                .toInstant().toEpochMilli()));
    }


}
